package com.bigdata.spark.flume;
import org.apache.commons.codec.binary.Base64;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author Gerry chan
 * @version 1.0
 * sparkSQL分析nginx日志
 * https://www.imooc.com/video/19726
 * 自定义flume过滤器：第一次数据清洗：格式化原始日志数据
 */
public class LogFormat implements Interceptor {

    private String date = null;
    private String year = "";
    private String month = "";
    private String day = "";
    private String hour = "";
    final Base64 base64 = new Base64();

    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        //获取Event信息头与信息体
        Map<String, String> head = event.getHeaders();
        String body = new String(event.getBody());
        //通过正则 提取时间
        Pattern p = Pattern.compile("\\[([^]]+)\\]");
        Matcher matcher = p.matcher(body);
        while (matcher.find()) {
            //将匹配到的数据提取出来
            String matcher1 = matcher.group(1);
            date = matcher1;
        }

        if (date != null) {
            //日期格式
            SimpleDateFormat sdf= new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss Z", Locale.ENGLISH);
           //提取日期
            SimpleDateFormat sd = new SimpleDateFormat("dd");
            //月份
            SimpleDateFormat sm = new SimpleDateFormat("MMM");
            //年份
            SimpleDateFormat sy = new SimpleDateFormat("yyyy");
            //小时
            SimpleDateFormat sh = new SimpleDateFormat("HH");

            try {
                Date _date = sdf.parse(date);
                year = sy.format(_date);
                month = sm.format(_date);
                day = sd.format(_date);
                hour = sh.format(date);
            } catch (ParseException e) {
                e.printStackTrace();
            }
            //提取埋点数据
            String[] _body = body.split(" ");
            String[] _split = _body[6].split("\\?");
            String mes = "";
            if (_split.length == 2) {
                //base64解密
                mes = new String(base64.decode(_split[1]));
            }
            // 将埋点数据 设置到body
            event.setBody(mes.getBytes());
            //设置提取的数据到信息头
            head.put("yearStr", year);
            head.put("monthStr", month);
            head.put("dayStr", day);
            head.put("hourStr", hour);
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        for(Event event:list) {
            intercept(event);
        }
        return list;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder{
        public Interceptor build() {
            return new LogFormat();
        }

        public void configure(Context context) {

        }
    }
}
